package com.easy.mq.client;

import java.util.Map;
import java.util.Properties;

import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.rocketmq.client.exception.MQClientException;

public final class KafkaProducerPool {

	public static Producer<String, String> getProducer(Map<String, Object> config) throws MQClientException {
	
		return  createProducerConfig(config);
	
	}


	private static Producer<String, String> createProducerConfig(Map<String, Object> config) {
		Properties props = new Properties();
		Producer<String, String> producer = null;
		for (Map.Entry<String, Object> pro : config.entrySet()) {
			if (StringUtils.isEmpty(String.valueOf(pro.getValue()))) {
				continue;
			}
			props.put(pro.getKey(), pro.getValue());
		}

		producer = new KafkaProducer<>(props);
		return producer;
	}

	public static void shutdown(Producer<String, String> producer) {
		if (producer != null) {
			try {
				producer.close();
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		
	}
}
